package com.sondertara.joya.core;


import com.sondertara.common.base.Assert;
import com.sondertara.common.collection.CollectionUtils;
import com.sondertara.common.collection.Lists;
import com.sondertara.common.collection.Maps;
import com.sondertara.common.collection.Sets;
import com.sondertara.common.exception.TaraException;
import com.sondertara.common.model.PageResult;
import com.sondertara.common.text.StringUtils;
import com.sondertara.common.timing.Stopwatch;
import com.sondertara.joya.core.constant.JoyaConfig;
import com.sondertara.joya.core.jdbc.callback.ConnectionCallback;
import com.sondertara.joya.core.jdbc.callback.TransactionCallback;
import com.sondertara.joya.core.jdbc.resultset.mapper.BeanRowMapper;
import com.sondertara.joya.core.jdbc.resultset.mapper.ListRecordExtractor;
import com.sondertara.joya.core.jdbc.resultset.mapper.MapRowMapper;
import com.sondertara.joya.core.jdbc.resultset.mapper.ResultSetExtractor;
import com.sondertara.joya.core.jdbc.resultset.mapper.RowMapper;
import com.sondertara.joya.core.jdbc.resultset.mapper.SingleColumnRowMapper;
import com.sondertara.joya.core.jdbc.resultset.mapper.SingleRowRecordExtractor;
import com.sondertara.joya.core.jdbc.resultset.mapper.StreamRecordExtractor;
import com.sondertara.joya.core.jdbc.statment.BatchPreparedStatementSetter;
import com.sondertara.joya.core.jdbc.statment.SimpleBatchPreparedStatementSetter;
import com.sondertara.joya.core.jdbc.statment.StatementCreatorUtils;
import com.sondertara.joya.core.jdbc.statment.creator.DeleteSqlCreator;
import com.sondertara.joya.core.jdbc.statment.creator.InsertSqlCreator;
import com.sondertara.joya.core.jdbc.statment.creator.SelectSqlCreator;
import com.sondertara.joya.core.jdbc.statment.creator.UpdateBatchByIdSqlCreator;
import com.sondertara.joya.core.jdbc.statment.creator.UpdateSqlCreator;
import com.sondertara.joya.core.jdbc.support.DataSourceUtils;
import com.sondertara.joya.core.jdbc.support.DbType;
import com.sondertara.joya.core.jdbc.support.JdbcAccessor;
import com.sondertara.joya.core.jdbc.support.JdbcType;
import com.sondertara.joya.core.jdbc.support.LoggingOutputStream;
import com.sondertara.joya.core.jdbc.support.ScriptRunner;
import com.sondertara.joya.core.jdbc.support.SqlParameterValue;
import com.sondertara.joya.core.jdbc.support.TableMetaContext;
import com.sondertara.joya.core.jdbc.support.dialect.BoundSql;
import com.sondertara.joya.core.jdbc.support.dialect.DialectFactory;
import com.sondertara.joya.core.jdbc.support.dialect.PageSqlDialect;
import com.sondertara.joya.core.jdbc.support.listener.EntityListener;
import com.sondertara.joya.core.jdbc.tran.TransactionTemplate;
import com.sondertara.joya.core.jdbc.tran.TransactionTemplateManager;
import com.sondertara.joya.core.model.EntityTableMetaDefinition;
import com.sondertara.joya.core.model.PersistEntity;
import com.sondertara.joya.core.query.NativeSqlQuery;
import com.sondertara.joya.core.query.criterion.WhereCondition;
import com.sondertara.joya.core.query.pagination.PageParam;
import com.sondertara.joya.exceptions.DbException;
import com.sondertara.joya.utils.JdbcUtils;
import com.sondertara.joya.utils.SqlUtils;
import org.jspecify.annotations.NonNull;
import org.jspecify.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;

import javax.sql.DataSource;
import java.io.PrintWriter;
import java.io.Reader;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.time.OffsetDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.UnaryOperator;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * JDBC工具类
 *
 * @author huangxiaohu
 */
public class JoyaJdbc extends JdbcAccessor implements JoyaOperations {

    private static final JoyaConfig GLOBAL_CONFIG = new JoyaConfig();
    private static final Logger logger = LoggerFactory.getLogger(JoyaJdbc.class);


    /**
     * 创建JdbcUtils
     *
     * @param dataSource 数据源
     */
    public JoyaJdbc(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    public static JoyaConfig config() {
        return GLOBAL_CONFIG;
    }

    /**
     * 自定义DataSource 销毁
     *
     * @param consumer func
     */
    public void destroy(Consumer<DataSource> consumer) {
        consumer.accept(this.dataSource);
    }


    /**
     * 自定义DataSource 销毁
     */
    public void destroy() {
        config().cacheClean();
    }

    /**
     * 判断当前是否在事务中
     */
    public boolean inTransaction() {
        return DataSourceUtils.getHolder(dataSource).hasTransaction();
    }

    /**
     * 查询数据库并转换结果集。
     * 用户可自定义结果集转换器。
     * 用户也可使用预定义的结果集转换器。
     *
     * @param sql          sql语句
     * @param recordMapper 结果集转换器
     * @param params       sql参数
     * @param <T>          resultSetMapper返回的结果类型
     * @return 成功则返回转换结果，失败则抛出DbException，结果为空则返回空列表
     * @see com.sondertara.joya.core.jdbc.resultset.mapper.ResultSetExtractor
     * @see ListRecordExtractor
     * @see SingleRowRecordExtractor
     */
    @Override
    public <T> T query(String sql, ResultSetExtractor<T> recordMapper, Object... params) {

        Stopwatch stopwatch = Stopwatch.createStarted();
        if (logger.isDebugEnabled()) {
            logger.debug("Joya#query start....");
        }
        return doWork(conn -> {
            try (PreparedStatement stmt = createPreparedStatement(conn, sql, false, params); ResultSet rs = stmt.executeQuery()) {
                return recordMapper.extract(rs);
            } finally {
                stopwatch.stop();
                if (logger.isDebugEnabled()) {
                    logger.debug("Joya#query end,cost {}....", stopwatch.stop().elapsed(TimeUnit.MILLISECONDS));
                }
            }
        });
    }

    /**
     * query stream
     *
     * @param sql       the sql
     * @param rowMapper row mapper
     * @param params    the query params
     * @param <T>       the return type
     * @return the stream
     */
    @Override
    public <T> Stream<T> queryStream(String sql, RowMapper<T> rowMapper, Object... params) {
        try {
            Connection connection = getConnection();
            PreparedStatement statement = createPreparedStatement(connection, sql, false, params);
            statement.setFetchSize(2000);
            ResultSet rs = statement.executeQuery();
            StreamRecordExtractor<T> recordMapper = gettStreamRecordMapper(rowMapper, rs, connection);
            return recordMapper.extract(rs);
        } catch (SQLException e) {
            throw new DbException(e);
        }

    }

    @NonNull
    private static <T> StreamRecordExtractor<T> gettStreamRecordMapper(RowMapper<T> rowMapper, ResultSet rs, Connection connection) {
        StreamRecordExtractor<T> recordMapper = new StreamRecordExtractor<>(rowMapper);
        recordMapper.delegateClose(() -> {
            try {
                if (null != rs) {
                    rs.close();
                }
            } catch (SQLException e) {
                throw new DbException(e);
            } finally {
                try {
                    connection.close();
                } catch (SQLException ignored) {
                }
            }

        });
        return recordMapper;
    }

    /**
     * 查询数据库，对结果集的每一行进行转换，然后将所有行封装成列表。
     * 用户可自定义行转换器。
     * 用户也可使用预定义的行转换器。
     *
     * @param sql       sql语句
     * @param rowMapper 行转换器
     * @param params    sql参数
     * @param <T>       rowMapper返回的结果类型
     * @return 成功则返回结果列表，失败则抛出DbException，结果为空则返回空列表
     * @see BeanRowMapper
     * @see MapRowMapper
     * @see SingleColumnRowMapper
     */
    @Override
    public <T> List<T> queryList(String sql, RowMapper<T> rowMapper, Object... params) {
        return query(sql, new ListRecordExtractor<>(rowMapper), params);
    }

    /**
     * 查询数据库，将结果集的每一行转换成JavaBean，然后将所有行封装成列表。
     *
     * @param sql    sql语句
     * @param type   JavaBean类型
     * @param params sql参数
     * @param <T>    JavaBean类型
     * @return 成功则返回结果列表，失败则抛出DbException，结果为空则返回空列表
     */
    @Override
    public <T> List<T> queryList(String sql, Class<T> type, Object... params) {
        return query(sql, new ListRecordExtractor<>(new BeanRowMapper<>(type)), params);
    }


    /**
     * 查询数据库，返回结果集中的单个值。
     * 如果结果集中有多个值，则只返回第一行第一列的值。
     *
     * @param sql    sql语句
     * @param params sql参数
     * @return 成功则返回结果值，失败则抛出DbException，结果为空则返回null
     */
    @Override
    public <T> T querySingleValue(String sql, Class<T> clazz, Object... params) {
        return query(sql, new SingleRowRecordExtractor<>(new SingleColumnRowMapper<>(clazz)), params);
    }


    /**
     * 查询数据库，返回结果集中的单行数据。
     * 如果结果集中有多行数据，则只返回第一行数据。
     * 用户可自定义行转换器。
     * 用户也可使用预定义的行转换器。
     *
     * @param sql       sql语句
     * @param rowMapper 行转换器
     * @param params    sql参数
     * @param <T>       rowMapper返回的结果类型
     * @return 成功则返回结果，失败则抛出DbException，结果为空则返回null
     * @see RowMapper
     * @see BeanRowMapper
     * @see MapRowMapper
     * @see SingleColumnRowMapper
     */
    @Override
    public <T> T querySingleRow(String sql, RowMapper<T> rowMapper, Object... params) {
        return query(sql, new SingleRowRecordExtractor<>(rowMapper), params);
    }

    /**
     * 查询数据库，将结果集中的单行数据转换成JavaBean。
     *
     * @param sql    sql语句
     * @param type   JavaBean类型
     * @param params sql参数
     * @param <T>    JavaBean类型
     * @return 成功则返回结果，失败则抛出DbException，结果为空则返回null
     */
    @Override
    public <T> T querySingleRow(String sql, Class<T> type, Object... params) {
        return querySingleRow(sql, new BeanRowMapper<>(type), params);
    }

    /**
     * 更新数据库，返回影响行数
     *
     * @param sql sql语句
     * @return 成功则返回影响行数，失败则抛出DbException
     */
    @Override
    public int execute(String sql) {
        return doWork(conn -> {
            try (Statement stmt = conn.createStatement()) {
                return stmt.executeUpdate(sql);
            }
        });
    }

    /**
     * 更新数据库，返回影响行数
     *
     * @param sql    sql语句
     * @param params sql参数
     * @return 成功则返回影响行数，失败则抛出DbException
     */
    @Override
    public int execute(String sql, Object[] params) {
        return doWork(conn -> {
            try (PreparedStatement stmt = createPreparedStatement(conn, sql, false, params)) {
                return stmt.executeUpdate();
            }
        });
    }


    @Override
    public Object insert(String sql, Object... params) {
        return doWork(conn -> {
            try (PreparedStatement stmt = createPreparedStatement(conn, sql, true, params)) {
                stmt.executeUpdate();
                ResultSet generatedKeys = stmt.getGeneratedKeys();
                if (generatedKeys.next()) {
                    return generatedKeys.getObject(1);
                }
                return null;
            }
        });
    }


    @Override
    public <T> void insertAll(List<T> dataList) {
        if (CollectionUtils.isEmpty(dataList)) {
            return;
        }
        List<TableMetaContext> insertRowsWithoutKey = new ArrayList<>();
        List<TableMetaContext> insertRowsWithKey = new ArrayList<>();


        for (T data : dataList) {
            TableMetaContext context = TableMetaContext.of(data);
            Object primaryKet = context.getPrimaryKeyValue();
            // auto increment
            if (primaryKet == null) {
                insertRowsWithoutKey.add(context);
            } else {
                insertRowsWithKey.add(context);
            }
        }
        String insertSqlWithoutKey;

        if (insertRowsWithoutKey.size() > 0) {
            insertSqlWithoutKey = buildInsertWithoutKey(insertRowsWithoutKey);

        } else {
            insertSqlWithoutKey = null;
        }
        String insertSqlWithKey;
        if (insertRowsWithKey.size() > 0) {
            insertSqlWithKey = buildInsertWithKey(insertRowsWithKey);
        } else {
            insertSqlWithKey = null;
        }
        logger.info("Batch insert start,insert-without-key size is:{},insert-with-key size is:{}", insertRowsWithoutKey.size(), insertRowsWithKey.size());
        if (logger.isDebugEnabled()) {
            logger.debug("Batch insert-without-key sql:{}", insertSqlWithoutKey);
            logger.debug("Batch insert-with-key sql:{}", insertSqlWithKey);
        }
        doWork(conn -> {
            batchUpdateInternal(insertSqlWithoutKey, insertRowsWithoutKey, conn, false);
            batchUpdateInternal(insertSqlWithKey, insertRowsWithKey, conn, false);
            return null;
        });

        logger.info("Batch update completed.");

    }

    @Override
    public <T> int updateByCondition(T entity, UnaryOperator<WhereCondition<T>> func) {
        TableMetaContext context = TableMetaContext.of(entity);
        UpdateSqlCreator<T> updateSqlCreator = new UpdateSqlCreator<>(context, true);
        updateSqlCreator.where(func);
        return execute(updateSqlCreator.create(), updateSqlCreator.createParameters());
    }

    @Override
    public <T> void updateIgnoreNull(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        context.propertyNotNull();
        context.initData();
        UpdateBatchByIdSqlCreator creator = new UpdateBatchByIdSqlCreator(context, true);
        execute(creator.create(), creator.createParameters());
    }

    @Override
    public <T> void update(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        context.initData();
        UpdateBatchByIdSqlCreator creator = new UpdateBatchByIdSqlCreator(context);
        execute(creator.create(), creator.createParameters());

    }

    public <T> T execute(String sql, PreparedStatementCallback<T> action) throws DbException {
        return execute(new SimplePreparedStatementCreator(sql), action, true);
    }

    /**
     * Determine SQL from potential provider object.
     *
     * @param obj object which is potentially an SqlProvider
     * @return the SQL string, or {@code null} if not known
     * @see SqlProvider
     */
    @Nullable
    private static String getSql(Object obj) {
        return (obj instanceof SqlProvider ? ((SqlProvider) obj).getSql() : null);
    }


    @Nullable
    public <T> T execute(PreparedStatementCreator psc, PreparedStatementCallback<T> action, boolean closeResources)
            throws DbException {

        Assert.notNull(psc, "PreparedStatementCreator must not be null");
        Assert.notNull(action, "Callback object must not be null");
        if (logger.isDebugEnabled()) {
            String sql = getSql(psc);
            logger.debug("Executing prepared SQL statement" + (sql != null ? " [" + sql + "]" : ""));
        }

        Connection con = obtainConnection();
        PreparedStatement ps = null;
        try {
            ps = psc.createPreparedStatement(con);
            return action.doInPreparedStatement(ps);
        } catch (SQLException ex) {
            JdbcUtils.closeStatement(ps);
            ps = null;
            DataSourceUtils.close(con);
            con = null;
            throw new DbException(ex.getCause());
        } finally {
            if (closeResources) {
                JdbcUtils.closeStatement(ps);
                DataSourceUtils.close(con);
            }
        }
    }


    private PreparedStatementCallback<int[]> getPreparedStatementCallback(BatchPreparedStatementSetter pss) {
        return ps -> {

            int batchSize = pss.getBatchSize();
            if (JdbcUtils.supportsBatchUpdates(ps.getConnection())) {
                for (int i = 0; i < batchSize; i++) {
                    pss.setValues(ps, i);
                    ps.addBatch();
                }
                return ps.executeBatch();
            } else {
                List<Integer> rowsAffected = new ArrayList<>();
                for (int i = 0; i < batchSize; i++) {
                    pss.setValues(ps, i);
                    rowsAffected.add(ps.executeUpdate());
                }
                int[] rowsAffectedArray = new int[rowsAffected.size()];
                for (int i = 0; i < rowsAffectedArray.length; i++) {
                    rowsAffectedArray[i] = rowsAffected.get(i);
                }
                return rowsAffectedArray;
            }
        };
    }


    @Override
    public int[] batchUpdate(String sql, final BatchPreparedStatementSetter pss) throws DbException {
        if (logger.isDebugEnabled()) {
            logger.debug("Executing SQL batch update [" + sql + "]");
        }
        int batchSize = pss.getBatchSize();
        if (batchSize == 0) {
            return new int[0];
        }

        int[] result = execute(sql, getPreparedStatementCallback(pss));
        Assert.state(result != null, "No result array");
        return result;
    }

    @Override
    public int[] batchUpdate(String sql, List<Object[]> batchArgs) {
        return batchUpdate(sql, batchArgs, new int[0]);
    }

    @Override
    public int[] batchUpdate(String sql, List<Object[]> batchArgs, int[] argTypes) {
        if (batchArgs.isEmpty()) {
            return new int[0];
        }
        return batchUpdate(
                sql,
                new BatchPreparedStatementSetter() {
                    @Override
                    public void setValues(PreparedStatement ps, int i) throws SQLException {
                        Object[] values = batchArgs.get(i);
                        int colIndex = 0;
                        for (Object value : values) {
                            colIndex++;
                            if (value instanceof SqlParameterValue) {
                                SqlParameterValue paramValue = (SqlParameterValue) value;
                                StatementCreatorUtils.setParameterValue(ps, colIndex, paramValue, paramValue.getValue());
                            } else {
                                int colType;
                                if (argTypes.length < colIndex) {
                                    colType = JdbcType.UNKNOWN.getCodes()[0];
                                } else {
                                    colType = argTypes[colIndex - 1];
                                }
                                StatementCreatorUtils.setParameterValue(ps, colIndex, colType, value);
                            }
                        }
                    }

                    @Override
                    public int getBatchSize() {
                        return batchArgs.size();
                    }
                });
    }

    @Override
    public <T> void updateAll(List<T> dataList) {
        if (CollectionUtils.isEmpty(dataList)) {
            return;
        }
        final List<TableMetaContext> updateList = new ArrayList<>();

        for (T data : dataList) {
            TableMetaContext context = TableMetaContext.of(data);
            updateList.add(context);

        }
        TableMetaContext context = updateList.get(0);
        UpdateBatchByIdSqlCreator creator = new UpdateBatchByIdSqlCreator(context);
        String updateSql = creator.create();
        updateList.forEach(metaContext -> {
            metaContext.handleEntityListener(listener -> listener.preUpdate(metaContext.getEntity()));
            metaContext.setPreparedParameterNames(creator.createParameterNames());
        });


        logger.info("Batch update start,update size is:{}", updateList.size());
        if (logger.isDebugEnabled()) {
            logger.debug("Batch update sql:{}", updateSql);
        }
        doWork(conn -> {
            batchUpdateInternal(updateSql, updateList, conn, true);
            return null;
        });
        logger.info("Batch update completed.");
    }


    /**
     * 开启事务
     */
    @Override
    public Connection startTransaction() {
        try {
            DataSourceUtils.getManager(this.dataSource).begin();
            return DataSourceUtils.getHolder(this.dataSource).getConnection();
        } catch (SQLException e) {
            throw new DbException(e);
        }

    }

    /**
     * 提交事务
     */
    @Override
    public void commit() {
        try {
            DataSourceUtils.getManager(this.dataSource).commit();
        } catch (SQLException e) {
            throw new DbException(e);
        }
    }

    /**
     * 回滚事务
     */
    @Override
    public void rollback() {
        try {
            DataSourceUtils.getManager(this.dataSource).rollBack();
        } catch (SQLException e) {
            throw new DbException(e);
        }
    }

    /**
     * run transaction with connection
     *
     * @param action callback
     * @param <T>    callback type
     * @return result
     */
    public <T> T doInTransactionCallback(TransactionCallback<T> action) {
        try {
            TransactionTemplate templateManager = new TransactionTemplateManager(DataSourceUtils.getManager(this.dataSource));
            return templateManager.execute(action);
        } catch (Throwable e) {
            throw new DbException(e);
        }

    }


    private Connection obtainConnection() {
        return DataSourceUtils.getConnection(this.dataSource);
    }

    public <T> T doWork(ConnectionCallback<T> action) {
        Connection conn = null;
        try {
            conn = obtainConnection();
            return action.doWork(conn);
        } catch (Exception e) {
            logger.error("Joya#doWork error,{}", e.getMessage(), e);
            throw new DbException(e);
        } finally {
            DataSourceUtils.close(conn);
        }
    }

    /**
     * run transaction with connection.
     *
     * @param action statement callback
     * @param <T>    type of result
     * @return result
     */
    public <T> T doInTransaction(ConnectionCallback<T> action) {

        try {
            return doInTransactionCallback(tranStatus -> {
                Connection connection = null;
                try {
                    connection = obtainConnection();
                    return action.doWork(connection);
                } finally {
                    DataSourceUtils.close(connection);
                }
            });
        } catch (Exception e) {
            logger.error("TransactionCallback", e);
            throw new TaraException(e);
        }
    }


    @Override
    public <T> T find(Class<T> tClass, UnaryOperator<WhereCondition<T>> func) {
        TableMetaContext context = TableMetaContext.of(tClass);
        SelectSqlCreator<T> creator = new SelectSqlCreator<>(context);
        if (null != func) {
            creator.where(func);
        }
        return querySingleRow(creator.create(), tClass, creator.createParameters());
    }

    @Override
    public <T> List<T> queryTop(String sql, Integer top, RowMapper<T> rowMapper, Object... params) {
        PageSqlDialect dialect = DialectFactory.getDialect(obtainConnection());
        BoundSql pageSql = dialect.pageSql(new BoundSql.BoundSqlObj(sql, params), 0, top);
        return queryList(pageSql.getSqlString(), rowMapper, pageSql.getArgs());

    }

    @Override
    public <T> List<T> queryTop(String sql, Integer top, Class<T> tClass, Object... params) {
        PageSqlDialect dialect = DialectFactory.getDialect(obtainConnection());
        BoundSql pageSql = dialect.pageSql(new BoundSql.BoundSqlObj(sql, params), 0, top);
        return queryList(pageSql.getSqlString(), tClass, pageSql.getArgs());

    }

    @Override
    public <T> List<T> findTop(Class<T> tClass, Integer topNum, UnaryOperator<WhereCondition<T>> func) {
        PageSqlDialect dialect = DialectFactory.getDialect(obtainConnection());
        TableMetaContext context = TableMetaContext.of(tClass);
        SelectSqlCreator<T> creator = new SelectSqlCreator<>(context);
        if (null != func) {
            creator.where(func);
        }
        BoundSql pageSql = dialect.pageSql(new BoundSql.BoundSqlObj(creator.create(), creator.createParameters()), 0, topNum);
        return queryList(pageSql.getSqlString(), tClass, pageSql.getArgs());

    }

    @Override
    public <T> List<T> findList(Class<T> tClass, UnaryOperator<WhereCondition<T>> func) {
        TableMetaContext context = TableMetaContext.of(tClass);
        SelectSqlCreator<T> creator = new SelectSqlCreator<>(context);
        if (null != func) {
            creator.where(func);
        }
        return queryList(creator.create(), tClass, creator.createParameters());
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T, R> List<R> findIdExists(Class<T> t, Collection<R> keys) {
        TableMetaContext context = TableMetaContext.of(t);
        SelectSqlCreator<Object> creator = new SelectSqlCreator<>(context);
        creator.setDeclaredColumns(Sets.newHashSet(context.getPrimaryKeyName()));
        creator.where(w -> w.in(context.getPrimaryKeyName(), keys));
        return (List<R>) queryList(creator.create(), context.getPrimaryKeyType(), creator.createParameters());
    }

    @Override
    public <T> List<T> findAllById(Class<T> t, Collection<?> keys) {
        TableMetaContext context = TableMetaContext.of(t);
        SelectSqlCreator<Object> creator = new SelectSqlCreator<>(context);
        creator.where(w -> w.in(context.getPrimaryKeyName(), keys));
        return queryList(creator.create(), t, creator.createParameters());
    }

    @Override
    public <T> T findById(Class<T> t, Serializable key) {
        TableMetaContext context = TableMetaContext.of(t);
        SelectSqlCreator<Object> creator = new SelectSqlCreator<>(context);
        creator.where(w -> w.eq(context.getPrimaryKeyName(), key));
        return querySingleRow(creator.create(), t, creator.createParameters());
    }


    @Override
    public <T> PageResult<T> queryPage(String sqlStr, RowMapper<T> rowMapper, PageParam page, Object... params) {
        return doWork(conn -> {
            PageSqlDialect dialect = DialectFactory.getDialect(conn);
            String jdbcStyleSql = SqlUtils.getJdbcStyleSql(sqlStr);
            BoundSql.BoundSqlObj boundSqlObj = new BoundSql.BoundSqlObj(jdbcStyleSql, params);
            BoundSql countSql = dialect.countSql(boundSqlObj);
            Long count = querySingleValue(countSql.getSqlString(), Long.class, countSql.getArgs());
            BoundSql pageSql = dialect.pageSql(boundSqlObj, page.startPosition(), page.getPageSize());
            List<T> list = queryList(pageSql.getSqlString(), rowMapper, pageSql.getArgs());
            return new PageResult<>(list, count, page.getPage(), page.getPageSize());
        });
    }

    @Override
    public <T> PageResult<T> queryPage(NativeSqlQuery nativeSqlQuery, RowMapper<T> rowMapper, PageParam page) {
        return queryPage(nativeSqlQuery.toSql(), rowMapper, page, nativeSqlQuery.getParams().toArray());
    }

    @Override
    public <T> PageResult<T> queryPage(NativeSqlQuery nativeSqlQuery, Class<T> tClass, PageParam page) {
        return queryPage(nativeSqlQuery, new BeanRowMapper<>(tClass), page);
    }

    @Override
    public <T> PageResult<T> queryPage(String sql, Class<T> tClass, PageParam page, Object... params) {
        return queryPage(sql, new BeanRowMapper<>(tClass), page, params);
    }


    @Override
    public <T> T find(NativeSqlQuery nativeSqlQuery, Class<T> tClass) {
        return querySingleRow(nativeSqlQuery.toSql(), tClass, nativeSqlQuery.getParams().toArray());
    }

    @Override
    public <T> List<T> findList(NativeSqlQuery nativeSqlQuery, Class<T> tClass) {
        return queryList(nativeSqlQuery.toSql(), tClass, nativeSqlQuery.getParams().toArray());
    }


    @Override
    public long count(NativeSqlQuery nativeSqlQuery) {
        String countSql = SqlUtils.buildCountSql(nativeSqlQuery.toSql());
        return querySingleValue(countSql, Long.class, nativeSqlQuery.getParams().toArray());
    }


    @Override
    @SuppressWarnings("unchecked")
    public <T> T query(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        SelectSqlCreator<T> creator = new SelectSqlCreator<>(context);
        WhereCondition<T> condition = new WhereCondition<>();
        for (Map.Entry<String, Object> entry : context.getData().entrySet()) {
            if (entry.getValue() != null) {
                condition.eq(entry.getKey(), entry.getValue());
            }
        }
        creator.setWhereCriterion(condition);
        return (T) querySingleRow(creator.create(), entity.getClass(), creator.createParameters());
    }

    @Override
    @SuppressWarnings("unchecked")
    public <T> List<T> findList(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        SelectSqlCreator<T> creator = new SelectSqlCreator<>(context);
        WhereCondition<T> condition = new WhereCondition<>();
        for (Map.Entry<String, Object> entry : context.getData().entrySet()) {
            if (entry.getValue() != null) {
                condition.eq(entry.getKey(), entry.getValue());
            }
        }
        creator.setWhereCriterion(condition);
        return (List<T>) queryList(creator.create(), entity.getClass(), creator.createParameters());
    }

    @Override
    public <T> int deleteById(Class<T> t, Serializable key) {
        TableMetaContext context = TableMetaContext.of(t);
        DeleteSqlCreator<T> creator = new DeleteSqlCreator<>(context);
        creator.where(w -> w.eq(context.getPrimaryKeyName(), key));
        return execute(creator.create(), creator.createParameters());
    }

    @Override
    public <T> int deleteByCondition(Class<T> t, UnaryOperator<WhereCondition<T>> func) {
        TableMetaContext context = TableMetaContext.of(t);
        DeleteSqlCreator<T> creator = new DeleteSqlCreator<>(context);
        creator.where(func);
        return execute(creator.create(), creator.createParameters());
    }

    @Override
    public <T> void delete(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        WhereCondition<T> criterion1 = new WhereCondition<>();
        for (Map.Entry<String, Object> entry : context.getData().entrySet()) {
            criterion1.eq(entry.getKey(), entry.getValue());
        }
        DeleteSqlCreator<T> creator = new DeleteSqlCreator<>(context);
        creator.setWhereCriterion(criterion1);
        execute(creator.create(), creator.createParameters());
    }

    @Override
    public <T> int deleteAllById(Class<T> clazz, Iterable<? extends Serializable> keys) {
        TableMetaContext context = TableMetaContext.of(clazz);
        DeleteSqlCreator<T> creator = new DeleteSqlCreator<>(context);
        creator.where(w -> w.in(context.getPrimaryKeyName(), Sets.newHashSet(keys)));
        return execute(creator.create(), creator.createParameters());
    }


    @Override
    public <T> Object insertIgnoreNull(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        InsertSqlCreator creator = new InsertSqlCreator(context, true);
        Object insert = insert(creator.create(), creator.createParameters());
        if (null == insert) {
            insert = context.getPrimaryKeyValue();
        }
        return insert;
    }

    @Override
    public <T> Object insert(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        InsertSqlCreator creator = new InsertSqlCreator(context);
        Object insert = insert(creator.create(), creator.createParameters());
        if (null == insert) {
            insert = context.getPrimaryKeyValue();
        }
        return insert;
    }


    @Override
    public <T> Object saveIgnoreNull(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        context.propertyNotNull();
        context.initData();
        return saveInternal(context, true);
    }

    @Override
    public <T> void saveAll(List<T> rows) {
        if (CollectionUtils.isEmpty(rows)) {
            return;
        }
        //超过8000行进行拆分
        List<List<T>> partitions = Lists.partition(rows, 8000);
        partitions.forEach(dataList -> {
            List<TableMetaContext> insertRowsWithoutKey = new ArrayList<>();
            List<TableMetaContext> insertRowsWithKey = new ArrayList<>();
            Map<Object, TableMetaContext> toUpdate = Maps.newLinkedHashMap();
            final List<TableMetaContext> updateList = new ArrayList<>();
            TableMetaContext one = TableMetaContext.of(dataList.get(0));
            for (T data : dataList) {
                TableMetaContext context = TableMetaContext.of(data);
                Object primaryKet = context.getPrimaryKeyValue();
                if (data instanceof PersistEntity) {
                    if (((PersistEntity) data).isNew()) {
                        insertRowsWithoutKey.add(context);
                    } else {
                        updateList.add(context);
                    }
                } else {
                    // auto increment
                    if (primaryKet == null) {
                        insertRowsWithoutKey.add(context);
                    } else {
                        toUpdate.put(primaryKet, context);
                    }
                }
            }
            if (!toUpdate.isEmpty()) {
                Set<Object> ids = toUpdate.keySet();
                List<Object> list = findIdExists(one.getMetaDefinition().getEntityClass(), ids);
                Set<Object> updates = ids.stream().filter(list::contains).collect(Collectors.toSet());
                toUpdate.forEach((key, v) -> {
                    if (updates.contains(key)) {
                        updateList.add(v);
                    } else {
                        insertRowsWithKey.add(v);
                    }
                });
                toUpdate.clear();
            }
            String insertSqlWithoutKey;


            if (insertRowsWithoutKey.isEmpty()) {
                insertSqlWithoutKey = null;
            } else {
                insertSqlWithoutKey = buildInsertWithoutKey(insertRowsWithoutKey);
            }
            String insertSqlWithKey;
            if (insertRowsWithKey.isEmpty()) {
                insertSqlWithKey = null;
            } else {
                insertSqlWithKey = buildInsertWithKey(insertRowsWithKey);
            }
            String updateSql;
            if (!updateList.isEmpty()) {
                TableMetaContext context = updateList.get(0);
                UpdateBatchByIdSqlCreator creator = new UpdateBatchByIdSqlCreator(context);
                updateSql = creator.create();
                updateList.forEach(metaContext -> {
                    metaContext.handleEntityListener(entityListener -> entityListener.preUpdate(metaContext.getEntity()));
                    metaContext.setPreparedParameterNames(creator.createParameterNames());
                });
            } else {
                updateSql = null;
            }

            logger.info("Batch update start,insert-without-key size is:{},insert-with-key size is:{},update size is:{}", insertRowsWithoutKey.size(), insertRowsWithKey.size(), updateList.size());
            if (logger.isDebugEnabled()) {
                logger.debug("Batch update sql:{}", updateSql);
                logger.debug("Batch insert-without-key sql:{}", insertSqlWithoutKey);
                logger.debug("Batch insert-with-key sql:{}", insertSqlWithKey);
            }

            doWork(conn -> {
                batchUpdateInternal(updateSql, updateList, conn, true);
                batchUpdateInternal(insertSqlWithoutKey, insertRowsWithoutKey, conn, false);
                batchUpdateInternal(insertSqlWithKey, insertRowsWithKey, conn, false);
                return null;
            });
        });
        logger.info("Batch update completed.");
    }

    private String buildInsertWithKey(List<TableMetaContext> insertRowsWithKey) {
        String insertSqlWithKey;
        TableMetaContext context = insertRowsWithKey.get(0);
        InsertSqlCreator insertSqlCreator = new InsertSqlCreator(context);
        insertSqlWithKey = insertSqlCreator.create();
        insertRowsWithKey.forEach(metaContext -> {
            metaContext.handleEntityListener(entityListener -> entityListener.prePersist(metaContext.getEntity()));
            metaContext.setPreparedParameterNames(insertSqlCreator.createParameterNames());
        });
        return insertSqlWithKey;
    }

    private String buildInsertWithoutKey(List<TableMetaContext> insertRowsWithoutKey) {
        String insertSqlWithoutKey;
        TableMetaContext context = insertRowsWithoutKey.get(0);
        InsertSqlCreator creator = new InsertSqlCreator(context);
        creator.setGeneratedKeyNames(Sets.newHashSet(context.getPrimaryKeyName()));
        insertSqlWithoutKey = creator.create();
        insertRowsWithoutKey.forEach(row -> {
            row.handleEntityListener(entityListener -> entityListener.prePersist(row.getEntity()));

            row.setPreparedParameterNames(creator.createParameterNames());
        });
        return insertSqlWithoutKey;
    }

    @Override
    public void close(Connection conn) {
        DataSourceUtils.close(conn);
    }

    @Override
    public void runScript(@NonNull Reader reader) {
        try (Connection connection = getConnection(); PrintWriter printWriter = new PrintWriter(new LoggingOutputStream(logger, Level.INFO)); PrintWriter printWriter1 = new PrintWriter(new LoggingOutputStream(logger, Level.ERROR))) {
            ScriptRunner scriptRunner = new ScriptRunner(connection);
            scriptRunner.setStopOnError(true);
            scriptRunner.setThrowWarning(true);
            scriptRunner.setLogWriter(printWriter);
            scriptRunner.setErrorLogWriter(printWriter1);
            scriptRunner.runScript(reader);
        } catch (Exception e) {
            throw new DbException(e);
        }

    }


    @Override
    public <T> Object save(T entity) {
        TableMetaContext context = TableMetaContext.of(entity);
        return saveInternal(context, false);
    }


    @Override
    public void dropTables(Class<?>... arr) throws SQLFeatureNotSupportedException {
        throw new SQLFeatureNotSupportedException("dropTables query is not supported yet.");
    }

    @Override
    public void createTables(Class<?>... arr) throws SQLFeatureNotSupportedException {
        throw new SQLFeatureNotSupportedException("createTables query is not supported yet.");
    }

    private Object saveInternal(TableMetaContext context, boolean skipNull) {
        Map<String, Object> data = context.getData();
        Object primaryKeyValue = context.getPrimaryKeyValue();
        String primaryKeyName = context.getPrimaryKeyName();
        EntityTableMetaDefinition metaDefinition = context.getMetaDefinition();
        if (null == primaryKeyValue) {
            InsertSqlCreator creator = new InsertSqlCreator(context, skipNull);
            return insert(creator.create(), creator.createParameters());
        }
        boolean entityListener = context.isEntityListener();

        Object value = querySingleValue(StringUtils.format("select {} from {} where {} = ?", primaryKeyName, context.getTableName(), primaryKeyName), Object.class, primaryKeyValue);

        if (null == value) {
            if (entityListener) {
                ((EntityListener) context.getEntity()).prePersist(context.getEntity());
            }

            InsertSqlCreator creator = new InsertSqlCreator(context, skipNull);
            insert(creator.create(), creator.createParameters());
            if (entityListener) {
                ((EntityListener) context.getEntity()).postPersist(context.getEntity());
            }
        } else {
            if (entityListener) {
                ((EntityListener) context.getEntity()).preUpdate(context.getEntity());
            }
            UpdateSqlCreator<?> creator = new UpdateSqlCreator<>(context, skipNull);
            creator.where(w -> w.eq(primaryKeyName, primaryKeyValue));
            execute(creator.create(), creator.createParameters());
            if (entityListener) {
                ((EntityListener) context.getEntity()).postUpdate(context.getEntity());
            }
        }
        return data.get(metaDefinition.getPrimaryKey());
    }

    private void batchUpdateInternal(String sql, List<TableMetaContext> rows, Connection connection, boolean isUpdate) throws SQLException {
        if (null == sql || rows.isEmpty()) {
            return;
        }
        boolean entityListener = rows.get(0).isEntityListener();
        SimpleBatchPreparedStatementSetter setter = new SimpleBatchPreparedStatementSetter(rows);
        PreparedStatement ps = connection.prepareStatement(sql);
        int batchSize = setter.getBatchSize();
        if (JdbcUtils.supportsBatchUpdates(ps.getConnection())) {
            for (int j = 0; j < batchSize; j++) {
                setter.setValues(ps, j);
                ps.addBatch();
                if ((j + 1) % JoyaJdbc.config().getBatchSize() == 0 || j == batchSize - 1) {
                    ps.executeBatch();
                    ps.clearBatch();
                    if (entityListener) {
                        List<TableMetaContext> list = rows.subList(Math.max(j + 1 - JoyaJdbc.config().getBatchSize(), 0), j + 1);
                        list.forEach(row -> {
                            if (isUpdate) {
                                ((EntityListener) row.getEntity()).postUpdate(row.getEntity());
                            } else {
                                ((EntityListener) row.getEntity()).postPersist(row.getEntity());
                            }
                        });
                    }
                }
            }
        } else {
            for (int i = 0; i < rows.size(); i++) {
                setter.setValues(ps, i);
                ps.executeUpdate();
                if (entityListener) {
                    if (isUpdate) {
                        ((EntityListener) rows.get(i).getEntity()).postUpdate(rows.get(i).getEntity());
                    } else {
                        ((EntityListener) rows.get(i).getEntity()).postPersist(rows.get(i).getEntity());
                    }
                }
            }
        }
        if (!ps.isClosed()) {
            ps.close();
        }
    }

    /**
     * 创建语句
     *
     * @param conn   连接
     * @param sql    sql语句
     * @param params sql参数
     * @return 创建的PreparedStatement对象
     * @throws SQLException 来自JDBC的异常
     */
    static PreparedStatement createPreparedStatement(Connection conn, String sql, boolean autoGeneratedKeys, Object... params) throws SQLException {
        PreparedStatement stmt;
        String jdbcStyleSql = SqlUtils.getJdbcStyleSql(sql);
        if (autoGeneratedKeys) {
            stmt = conn.prepareStatement(jdbcStyleSql, Statement.RETURN_GENERATED_KEYS);
        } else {
            stmt = conn.prepareStatement(jdbcStyleSql);
        }
        if (null != params && params.length > 0) {
            for (int i = 0; i < params.length; i++) {
                if (params[i] instanceof java.sql.Date) {
                    stmt.setDate(i + 1, (java.sql.Date) params[i]);
                } else if (params[i] instanceof java.sql.Timestamp) {
                    stmt.setTimestamp(i + 1, (java.sql.Timestamp) params[i]);
                } else if (params[i] instanceof Date) {
                    if (DbType.oracle.equals(DialectFactory.getDatabaseType(conn))) {
                        stmt.setTimestamp(i + 1, Timestamp.from(((Date) params[i]).toInstant()));
                    } else {
                        stmt.setTimestamp(i + 1, Timestamp.from(((Date) params[i]).toInstant()));
                    }
                } else if (params[i] instanceof OffsetDateTime) {
                    if (DbType.oracle.equals(DialectFactory.getDatabaseType(conn))) {
                        OffsetDateTime offsetDateTime = (OffsetDateTime) params[i];
                        if (offsetDateTime.getYear() < 1905) {
                            stmt.setTimestamp(i + 1, Timestamp.valueOf(offsetDateTime.atZoneSameInstant(ZoneId.systemDefault()).toLocalDateTime()));
                        } else {
                            stmt.setTimestamp(i + 1, Timestamp.from(offsetDateTime.toInstant()));
                        }
                    } else {
                        stmt.setObject(i + 1, params[i]);
                    }
                } else {
                    stmt.setObject(i + 1, params[i]);
                }
            }
        }
        return stmt;
    }

    /**
     * Simple adapter for PreparedStatementCreator, allowing to use a plain SQL statement.
     */
    private static class SimplePreparedStatementCreator implements PreparedStatementCreator, SqlProvider {

        private final String sql;

        public SimplePreparedStatementCreator(String sql) {
            Assert.notNull(sql, "SQL must not be null");
            this.sql = sql;
        }

        @Override
        public PreparedStatement createPreparedStatement(Connection con) throws SQLException {
            return con.prepareStatement(this.sql);
        }

        @Override
        public String getSql() {
            return this.sql;
        }
    }


}
